// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: xguo

#include "coin2/exchange/base/api_base/asio_http_driver.h"

#include <iostream>
#include <memory>
#include <string>

#include "coin2/exchange/base/api_base/asio_http_context.h"
#include "coin2/exchange/base/log/latency_recorder.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/order/connection.pb.h"

namespace coin2::exchange::base::api_base {

using LatencyRecorder = coin2::exchange::base::strategy_util::LatencyRecorder;
using HttpDriverManager = coin2::exchange::base::api_base::HttpDriverManager;
using OrderSubTopicId = coin2::exchange::base::order::OrderSubTopicId;

NativeHttpAsioDriver::NativeHttpAsioDriver(
    boost::asio::io_context* ioc,
    const MarketExchangeApi& mea,
    const OrderConnectionConfig& cfg)
    : pool_(HttpDriverManager::Get(mea, cfg, ioc, cfg.max_sessions_per_host())) {
  LOG(INFO) << "max_sessions_per_host: " << cfg.max_sessions_per_host();
  if (cfg.http2_as_backup_http_driver() && mea.exchange == ExchangeType::Binance) {
    OrderConnectionConfig tmp;
    tmp.MergeFrom(cfg);
    tmp.set_use_http2(true);
    http2_pool_ = HttpDriverManager::Get(mea, tmp, ioc);
  }
}

bool NativeHttpAsioDriver::is_blocking() const {
  is_pool_blocking_ = pool_->is_blocking();
  return !http2_pool_ && is_pool_blocking_;
}

bool NativeHttpAsioDriver::SendByHttp2(const std::shared_ptr<AsioHttpContext>& context) const {
  if (!http2_pool_) {
    return false;
  }
  if (is_pool_blocking_) {
    return true;
  }
  if (context->query_type == OrderSubTopicId::REST_ALL_ORDERS) {  // query history order via http2
    return true;
  }

  return false;
}

void NativeHttpAsioDriver::Start() {
  pool_->SetTimeout(5'000'000'000LL);
  pool_->StartTimer();
  LOG(INFO) << "Default timeout 5 seconds";
}

void NativeHttpAsioDriver::Stop() { pool_->StopTimer(); }

void NativeHttpAsioDriver::PostAsyncContext(const std::shared_ptr<AsioHttpContext>& context) {
  int64_t ts = GetCurrentTimestamp();
  context->SetStatus(AsioHttpContextStatus::kRequested, ts);
  if (!context->keep_user_agent) {
    context->req.erase(boost::beast::http::field::user_agent);
  }
  auto* pool = SendByHttp2(context) ? http2_pool_.get() : pool_.get();
  pool->AsyncRequest(
      context->host,
      std::stoi(context->port),
      context->bind_ip,
      context->req,
      std::bind(
          &NativeHttpAsioDriver::ProcessResponse,
          this,
          context,
          ts,
          std::placeholders::_1,
          std::placeholders::_2),
      GetPriorityWeight(context->query_type, context->weight));
  if (latency_recorder_ != nullptr) {
    DLOG(INFO) << "debug TPS stats: " << GetTps().DebugString();
    if (IsReqWindowFull(ts)) {
      latency_recorder_->WriteTps(GetTps());
      request_window_.Clear();
      last_emit_time_ = ts;
    }
  }
  request_window_.Push(ts, 1);
  LOG_EVERY_N(INFO, 100) << "TPS stats: " << GetTps().DebugString();
}

void NativeHttpAsioDriver::set_latency_recorder(LatencyRecorder* latency_recorder) {
  latency_recorder_ = latency_recorder;
}

void NativeHttpAsioDriver::ProcessResponse(
    const std::shared_ptr<AsioHttpContext>& context,
    int64_t request_id,
    const boost::system::error_code& ec,
    const SessionResponse* response) {
  context->error_code.set_error_code(ec.value());
  context->error_code.set_error_message(ec.message());
  context->error_code.set_category(ec.category().name());

  context->response_timestamp = response->response_ts;
  context->request_timestamp = response->request_ts;
  if (response->raw_response) {
    context->res = *(response->raw_response);
  }
  if (ec.value() == boost::system::errc::success) {
    context->SetStatus(AsioHttpContextStatus::kResponded, response->response_ts);
  } else {
    if (ec.value() == boost::system::errc::timed_out) {
      context->SetStatus(AsioHttpContextStatus::kTimeout, response->response_ts);
    } else {
      context->SetStatus(AsioHttpContextStatus::kError, response->response_ts);
    }
    if (ec.value() == boost::system::errc::too_many_files_open) {
      context->res.result(420 /* CUSTOMIZED: NO AVAILABLE CONNECTION */);
    } else {
      SPDLOG_ERROR("Http Request Error: \n{}", context->ToProtoShort().DebugString());
    }
  }
  context->handler(context);
}

bool NativeHttpAsioDriver::IsReqWindowFull(int64_t ts) const {
  return ts - last_emit_time_ > window_size_;
}

auto NativeHttpAsioDriver::GetTps() const -> const TpsProto& {
  tps_proto_.Clear();
  tps_proto_.set_window_size(request_window_.time_range());
  tps_proto_.set_num_reqs(request_window_.sum());
  return tps_proto_;
}

void NativeHttpAsioDriver::ClearReqWindow() { request_window_.Clear(); }
auto NativeHttpAsioDriver::GetPriorityWeight(int query_type, AsioHttpContextWeight orig_weight)
    const -> AsioHttpContextWeight {
  auto weight = orig_weight;
  using coin::proto::OrderSubTopicId;
  auto topic = static_cast<OrderSubTopicId>(query_type);
  switch (topic) {
    case OrderSubTopicId::SUBMIT_RESPONSE:
    case OrderSubTopicId::AMEND_RESPONSE:
    case OrderSubTopicId::CANCEL_RESPONSE:
    case OrderSubTopicId::MULTI_SUBMIT_RESPONSE:
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      weight = AsioHttpContextWeight::kHigh;
      break;
    case OrderSubTopicId::LISTEN_KEY:
      weight = AsioHttpContextWeight::kCritical;
      break;
    default:
      break;
  }

  return weight;
}

}  // namespace coin2::exchange::base::api_base
